package com.example.flow;

import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * “线程复用”不等于“永远同一个线程处理所有回调”
 * 线程复用的意思是：不会像阻塞模式那样，每个请求/任务都固定占用一个独立线程直到完成；而是任务在不同时间片中复用有限数量的线程（比如 Netty 的事件循环线程，或 ForkJoinPool 里的工作线程）。
 * 在 响应式流（Reactive Streams） 规范里，并没有要求所有 onNext() 必须在同一个线程执行；只是规定同一时间不会并发调用同一个 Subscriber 的多个信号（串行保证）。
 * SubmissionPublisher 默认用的是 ForkJoinPool.commonPool，它是一个多线程线程池。你的每次数据投递（submit()）都可能被不同的工作线程取出来回调 onNext()。
 * 所以：
 * 在一个批次里，可能是同一个线程连续调用 onNext()（看起来像“一个线程不断处理数据”）。
 * 但批次与批次之间，可能换了另一个线程来调用后续的 onNext()。
 */
public class FlowDemoFixed {

    public static void main(String[] args) {

        /**
         * 发布者
         * SubmissionPublisher 的实现是多线程的（默认用 ForkJoinPool.commonPool），而且有内部队列。
         * 你 request(5) 后的执行方式：
         * 发布者会根据配额，最多取出 5 条数据。
         * 每条数据的 onNext() 可能被分配给线程池中的不同线程去调用，但会用同步机制保证同一时刻只有一个 onNext() 在执行（串行）。
         * 如果上一次 onNext() 是线程 A 调的，下一次可能就是线程 B 调的，但它会等线程 A 的调用结束后才开始。
         * 原因：
         * SubmissionPublisher 对每个订阅者有一个 Executor 任务队列，确保信号顺序；
         * 投递动作是线程池里的任务，不同任务可能被不同线程取走执行。
         */
        SubmissionPublisher<String> publisher = new SubmissionPublisher<>();

        /**
         * 订阅者
         */
        Flow.Subscriber<String> subscriber = new Flow.Subscriber<String>() {
            /**
             * 订阅关系
             */
            private Flow.Subscription subscription;
            /**
             * 订阅数量
             */
            private int count = 0;
            /**
             * 是否完成
             * 并发可见性与原子性：为什么要用 AtomicBoolean/volatile
             * 场景：取消后仍可能回调 onNext 1～几次。我们需要跨线程共享“已取消”状态，且保证只取消一次。
             * 可见性（Visibility）
             * 含义：一个线程写入的最新值，其他线程能立刻看到。
             * 问题：普通字段可能被 CPU 缓存“滞留”，读到旧值。
             * 解决：volatile 或 Atomic*（内部含 volatile 语义）。
             * 原子性（Atomicity）
             * 含义：一个复合操作“不可分割”，不被其他线程插队。
             * 问题：多线程“先读再改”会竞争，可能多次执行“只应执行一次”的逻辑。
             * 解决：synchronized/Lock 或 AtomicBoolean.compareAndSet(false, true)（CAS）。
             * 在响应式里的用途
             * 闸门：在 onNext 首行检查 cancelled（可见性），为真就立刻返回；
             * 只取消一次：首次命中取消条件时 CAS 把 false→true（原子性），避免并发重复取消/重复清理。
             * 易错点：把 AtomicBoolean 当“锁”。不是的！它只是一个原子变量，不阻塞也不串行化线程；回调仍然可能换线程，只是有了“闸门”，尾巴被快速丢弃。
             */
            private final AtomicBoolean isCompleted = new AtomicBoolean(false);

            /**
             * Reactive Streams 规范的要求
             * 串行性保证：
             * 规范规定，对同一个 Subscriber 来说，onNext() 的调用不能并发重入，必须按顺序调用。
             * 也就是说，在一个 onNext() 没返回之前，不能开始另一个 onNext() 调用。
             * 这是为了保证你在回调里可以安全地维护本地状态（比如计数器）而不用加锁。
             * 线程一致性没有强制要求：
             * 规范并没有要求所有 onNext() 必须在同一个线程执行。它只要求“按顺序”，但允许换线程执行下一次回调。
             * @param subscription a new subscription
             */

            @Override
            public void onSubscribe(Flow.Subscription subscription) {
                this.subscription = subscription;
                /**
                 * 初次申请 5 个
                 */
                this.subscription.request(5);
            }

            @Override
            public void onNext(String item) {
                /**
                 * 若已取消，直接丢弃剩余回调（竞态保护）
                 */
                if (isCompleted.get()) {
                    return;
                }
                /**
                 * 命中阈值就立即取消并返回，避免后续任何处理/请求
                 */
                if (count == 3803) {
                    isCompleted.set(true);
                    this.subscription.cancel();
                    System.out.println("取消订阅");
                    return;
                }
                System.out.println("订阅数据：" + item);
                count++;
                /**
                 * 每处理满 5 个，再申请 5 个
                 */
                if (count % 5 == 0 && !isCompleted.get()) {
                    this.subscription.request(5);
                }
            }

            @Override
            public void onError(Throwable throwable) {
                System.out.println("订阅数据异常：" + throwable);
            }

            @Override
            public void onComplete() {
                System.out.println("订阅数据完成");
            }
        };

        /**
         * 发布者绑定订阅者
         */
        publisher.subscribe(subscriber);

        /**
         * 发布者发布数据
         */
        for (int i = 0; i < 5000; i++) {
            publisher.submit("数据：" + i);
        }

        /**
         * 关闭发布者
         */
        publisher.close();

        /**
         * 休眠 5 秒，等待数据处理完成
         */
        try {
            Thread.sleep(5000);
        } catch (InterruptedException e) {
            System.out.println(e.getMessage());
        }
        System.out.println("数据处理完成");

    }

}
/**
 * 串行但可换线程 —— 既安全又高利用
 * 规范规定：对同一个订阅者的 onNext/onError/onComplete 不能并发重入（保证串行性）；
 * 但不要求全程同一线程。SubmissionPublisher 默认用 ForkJoinPool.commonPool，不同回调可在不同线程顺序执行。
 * 这正是“线程复用”的要义：少量线程服务大量事件，而不是阻塞地长期占用。
 */